package com.study.chapter06;

import com.study.entity.Event;
import com.study.chapter05.source.ClickSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/3 13:31
 */
public class WindowTest {
    public static void main(String[] args) throws Exception {
        //环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> stream = env.addSource(new ClickSource());

        WindowedStream<Event, String, TimeWindow> window = stream.keyBy(data -> data.user)
                //滑动处理时间窗口
                // .window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))
                //滚动事件时间窗口
                // .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                //处理时间会话窗口
                // .window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)))
                //计数窗口
                // .countWindow(10)
                //滚动处理时间窗口
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        stream.print();

        env.execute();
    }
}
